package my.hadoop.mapreduce.example.sort;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Sort extends Configured implements Tool {

	public static class Map extends
			Mapper<LongWritable, Text, IntWritable, IntWritable> {

		private static IntWritable data = new IntWritable();
		private static IntWritable one = new IntWritable(1);

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			data.set(Integer.parseInt(value.toString()));
			context.write(data, one);
		}
	}

	public static class Reduce extends
			Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {

		private static IntWritable linenum = new IntWritable(1);

		@Override
		public void reduce(IntWritable key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			for (IntWritable value : values) {
				context.write(linenum, key);
				linenum.set(linenum.get() + 1);
			}
		}
	}

	public static class Partition extends Partitioner<IntWritable, IntWritable> {

		@Override
		public int getPartition(IntWritable key, IntWritable value, int num) {
			int Maxnumber = 65223;
			int bound = Maxnumber / num + 1;
			int keynumber = key.get();

			for (int i = 0; i < num; i++) {
				if (keynumber < bound * (i + 1) && keynumber >= bound * i) {
					return i;
				}
			}

			return -1;
		}

	}

	public static void main(String[] args) throws Exception {
		int status = ToolRunner.run(new Configuration(), new Sort(), args);
		System.exit(status);
	}

	public int run(String[] args) throws Exception {
		Job job = Job.getInstance(getConf(), "Sort");
		job.setJarByClass(getClass());
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);

		job.setPartitionerClass(Partition.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		return job.waitForCompletion(true) ? 0 : 1;
	}

}
